package com.autonavi.collapse.task;

import com.autonavi.collapse.concurrent.AsyncScheduledThreadPool;
import com.autonavi.collapse.concurrent.AsyncThreadPool;
import com.autonavi.collapse.concurrent.IdentifyThreadFactory;
import com.autonavi.collapse.config.ConfigManager;
import com.autonavi.collapse.config.StaticConfigManager;
import com.autonavi.collapse.task.model.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class TaskRuntimeManager implements TaskListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(TaskRuntimeManager.class);

    private TaskExecutorContainer taskExecutorContainer;

    private TaskService taskService;

    private ConfigManager configManager;

    private ExecutorService executorService;

    private ScheduledExecutorService heartbeatExecutorService;

    private int poolSize = Runtime.getRuntime().availableProcessors();

    private String myName;
    private int heartbeatTimeout = 30;
    private int heartbeatInterval = 5;

    public TaskRuntimeManager(String myName){
        this.myName = myName;
    }

    @Override
    public void init() throws Exception {
        executorService = new AsyncThreadPool(poolSize, new IdentifyThreadFactory("task"));
        heartbeatExecutorService = new AsyncScheduledThreadPool(1);
        heartbeatExecutorService.scheduleAtFixedRate(new HeartbeatTask(), heartbeatInterval, heartbeatInterval, TimeUnit.SECONDS);
        LOGGER.info("Node heartbeat statrt");
        if (configManager == null) {
            try {
                configManager = new StaticConfigManager();
                configManager.init();
            }catch (Exception e) {
                LOGGER.warn(e.getMessage());
            }
        }
        if (taskExecutorContainer == null) {
            TaskExecutorManager taskExecutorManager = new TaskExecutorManager();
            taskExecutorManager.init();
            taskExecutorContainer = new TaskExecutorContainer();
            taskExecutorContainer.setTaskExectorManager(taskExecutorManager);
        }
    }

    @Override
    public void destroy() {
        heartbeatExecutorService.shutdown();
        executorService.shutdown();
    }

    @Override
    public List<TaskConfig> onCreate(TaskTemplate taskTemplate) {
        ExtendedMap extendedMap = configManager.findConfig(taskTemplate.getName());
        Map<String, ConfigValue> configs;
        if (extendedMap != null) {
            configs = extendedMap.getAll();
            Map<String, ConfigValue> remoteConfigs = taskTemplate.getConfigs();
            if(remoteConfigs != null){
                configs.putAll(remoteConfigs);
            }
            taskTemplate.setConfigs(configs);
        }
        return taskExecutorContainer.create(taskTemplate);
    }

    @Override
    public void onDispatch(TaskConfig task) {
        String id = task.getId();
        if (taskExecutorContainer.isRunning(id)) {
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("Task {} is running " , id);
            }
            return;
        }
        if (!taskService.setDispatched(id, myName)) {
            LOGGER.warn("Task {} set to dispatch fail " , id);
            return;
        }
        TaskContext context = new TaskContext(task);
        executorService.execute(() -> {
            if (taskService.setRunning(id, myName)) {
                LOGGER.info("Task {} start", id);
                TaskResult taskResult = null;
                try {
                    taskResult = taskExecutorContainer.execute(task.getType(),context);
                }catch (Exception e) {
                    LOGGER.error(e.getMessage(),e);
                    taskResult = taskResult.fail(task, e);
                }finally {
                    LOGGER.info("Task {} finish, status is {} ", id, taskResult);
                }
                try {
                    taskService.notify(taskResult, myName);
                }catch (Exception e) {
                    LOGGER.error(e.getMessage(), e);
                }
            }
        });
    }

    @Override
    public void onCancel(String taskId) {
        try {
            taskExecutorContainer.interrupt(taskId);
        }catch (InterruptedException e) {
            LOGGER.error(e.getMessage());
        }
    }

    public void setTaskExecutorContainer(TaskExecutorContainer taskExecutorContainer) {
        this.taskExecutorContainer = taskExecutorContainer;
    }

    public void setTaskService(TaskService taskService) {
        this.taskService = taskService;
    }

    public void setHeartbeatTimeout(int heartbeatTimeout) {
        this.heartbeatTimeout = heartbeatTimeout;
    }

    public void setHeartbeatInterval(int heartbeatInterval) {
        this.heartbeatInterval = heartbeatInterval;
    }

    private class HeartbeatTask implements Runnable {

        @Override
        public void run() {
            int freeSize = poolSize - taskExecutorContainer.getRunningTaskSize();
            if (freeSize < 0) {
                freeSize = 0;
            }
            taskService.heartbeat(myName, freeSize, heartbeatTimeout);
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("{} heartbeat ", myName);
            }
        }
    }
}
